package org.budo.canal.server;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;

import org.budo.canal.instance.spring.CanalInstanceFactoryBean;
import org.budo.canal.message.handler.CanalMessageHandler;
import org.budo.support.lang.util.CollectionUtil;
import org.budo.support.lang.util.ThreadUtil;
import org.budo.support.spring.util.SpringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitor;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitors;
import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * 会自己去IOC里找ZkClientx; 调用 start 方法启动后，会初始化并扫描所有 CanalInstance 并启动监听
 * 
 * @author limingwei
 */
@Getter
@Setter
@Slf4j
public class BudoCanalServerImpl extends AbstractBudoCanalServer {
    private static final Logger threadLog = LoggerFactory.getLogger("BudoCanalServer_getWithoutAck_thred");

    public void start() {
        try {
            this.startDestinations();
        } catch (Throwable e) {
            log.error("#36 canalServer.start error, e=" + e, e);
        }
    }

    /**
     * 启动所有 CanalInstance
     */
    private void startDestinations() {
        String[] instanceNames = this.getApplicationContext().getBeanNamesForType(CanalInstance.class);
        log.info("#42 startDestinations, instanceNames=" + Arrays.toString(instanceNames));

        for (String beanName : instanceNames) {
            String destination = beanName; // 以 beanName 作为 destination
            try {
                this.startDestination(destination);
            } catch (Exception e) { // 一个失败后，启动下一个
                log.error("#49 startGetWithoutAckThread_1 for " + destination + " error, e=" + e, e);
            }
        }
    }

    protected void onServerRunningMonitorProcessActiveEnter(final CanalServerWithEmbedded canalServer, final String destination) {
        log.info("#72 serverRunningMonitorProcessActiveEnter, start destination " + destination);

        canalServer.start(destination);
        this.startDestination(destination);
    }

    /**
     * 启动一个 CanalInstance
     */
    public void startDestination(String destination) {
        CanalServerWithEmbedded canalServer = this.canalServer();
        log.info("#66 startGetWithoutAckThread_1, canalServer=" + canalServer + ", destination=" + destination);

        if (!canalServer.isStart(destination)) {
            ServerRunningMonitor serverRunningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
            log.info("#69 serverRunningMonitor.start, destination=" + destination);
            serverRunningMonitor.start(); // 如果数据库连接不上，这里会报错
        }

        // 二次检查，如果多台负载均衡的话，可能不会启动
        if (!canalServer.isStart(destination)) {
            log.error("#77 destination=" + destination + ", not started, canalServer=" + canalServer);
            return;
        }

        short clientId = 1001; // @see https://github.com/alibaba/canal/wiki/ClientAPI
        ClientIdentity clientIdentity = new ClientIdentity(destination, clientId);

        canalServer.subscribe(clientIdentity); // subscribe 应可重复调用时
                                               // @seecom.alibaba.otter.canal.meta.MemoryMetaManager#subscribe(ClientIdentity)
        log.info("#68 after subscribe, canalServer=" + canalServer + ", clientIdentity=" + clientIdentity);

        this.startGetWithoutAckThread_2(canalServer, clientIdentity);
    }

    private void startGetWithoutAckThread_2(final CanalServerWithEmbedded canalServer, final ClientIdentity clientIdentity) {
        String threadName = "CanalGetThread-" + clientIdentity.getDestination();
        Thread thread = ThreadUtil.getThreadByName(threadName);
        if (null != thread) { // 重复调用start时判断线程是否已经存在
            log.warn("#97 thread exists, thread=" + thread + ", clientIdentity=" + clientIdentity + ", canalServer=" + canalServer);
            return;
        }

        log.info("#101 startGetWithoutAckThread_2, threadName=" + threadName + ", canalServer=" + canalServer + ", clientIdentity=" + clientIdentity);

        Runnable target = new Runnable() {
            public void run() {
                log.info("#105 startGetWithoutAckThread_2, Thread started, canalServer=" + canalServer + ", clientIdentity=" + clientIdentity + ", thread=" + this);

                while (true) {
                    BudoCanalServerImpl.this.getWithoutAck(canalServer, clientIdentity);
                }
            }
        };

        Thread newThread = new Thread(target, threadName);
        newThread.setDaemon(true);
        newThread.start();
    }

    /**
     * 拉取 binlog
     */
    private void getWithoutAck(CanalServerWithEmbedded canalServer, ClientIdentity clientIdentity) {
        Message message = null;
        try {
            CanalInstanceFactoryBean canalInstanceFactoryBean = (CanalInstanceFactoryBean) SpringUtil.getBean(this.getApplicationContext(), "&" + clientIdentity.getDestination());

            Integer batchSize = canalInstanceFactoryBean.getBatchSize();
            Long timeout = canalInstanceFactoryBean.getTimeout();
            Long interval = canalInstanceFactoryBean.getInterval();
            CanalMessageHandler canalMessageHandler = canalInstanceFactoryBean.getCanalMessageHandler();

            message = canalServer.getWithoutAck(clientIdentity, batchSize, timeout, TimeUnit.MILLISECONDS);
            if (message == null || message.getId() == -1 || CollectionUtil.isEmpty(message.getEntries())) {
                threadLog.debug("#124 message={}, sleep {}, clientIdentity={}, canalInstanceFactoryBean={}", message, interval, clientIdentity, canalInstanceFactoryBean);

                ThreadUtil.sleep(interval);
                return;
            }

            canalMessageHandler.handleMessage(message, clientIdentity);
            canalServer.ack(clientIdentity, message.getId()); // ack
        } catch (Throwable e) {
            log.error("#132 getWithoutAck error, rollback now, e=" + e, e);

            if (null != message) {
                canalServer.rollback(clientIdentity, message.getId()); // rollback
                log.info("#136 rollback successfully, clientIdentity={}, message={}", clientIdentity, message);
            }
        }
    }
}